// SPDX-License-Identifier: Mulan PSL v2
/*
 * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 * This software is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *         http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

use std::{
    ops::{Deref, DerefMut},
    time::Duration,
};

use super::{
    bytewise::{BytewiseBuffer, BytewiseReadOwned, BytewiseWrite, BytewiseWriter},
    framer::Framer,
    message::{Request, Response},
    transport::{Endpoint, Transport},
    Error,
};

#[derive(Debug)]
pub struct Peer {
    framer: Framer,
    endpoint: Endpoint,
}

impl Peer {
    #[inline]
    pub fn new(framer: Framer, endpoint: Endpoint) -> Self {
        Self { framer, endpoint }
    }

    pub fn send_message<B: BytewiseWrite>(&self, message: &B) -> Result<(), Error> {
        let mut write_buf = self.endpoint.write()?;
        let mut frame_buf = self.framer.encode_frame(&mut write_buf)?;

        let mut writer = BytewiseBuffer::new(frame_buf.as_mut());
        message.write_to(&mut writer)?;

        let payload_len = writer.written_bytes();
        let frame_len = frame_buf.finalize(payload_len)?;

        write_buf.submit(frame_len)?;

        Ok(())
    }

    pub fn receive_message<B: BytewiseReadOwned>(&self) -> Result<Option<B>, Error> {
        let read_buf = self.endpoint.read()?;

        let frame = match self.framer.decode_frame(&read_buf)? {
            Some(result) => result,
            None => return Ok(None),
        };

        let mut reader = BytewiseBuffer::new(frame.as_ref());
        let message = B::read_from_mut(&mut reader)?;

        let frame_len = frame.frame_len();

        read_buf.consume(frame_len)?;

        Ok(Some(message))
    }

    pub fn invoke(&self, request: &Request) -> Result<Response<'_>, Error> {
        self.send_message(request)?;

        let response = loop {
            match self.receive_message()? {
                Some(resp) => break resp,
                None => continue,
            }
        };

        Ok(response)
    }
}

#[repr(transparent)]
#[derive(Debug)]
pub struct Server(Peer);

impl Server {
    #[inline]
    pub fn create<S: AsRef<str>>(
        transport: &Transport,
        framer: Framer,
        name: S,
    ) -> Result<Self, Error> {
        let endpoint = transport.create(name).map_err(Error::TransportError)?;

        Ok(Self(Peer::new(framer, endpoint)))
    }
}

impl Deref for Server {
    type Target = Peer;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl DerefMut for Server {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}

#[repr(transparent)]
#[derive(Debug)]
pub struct Client(Peer);

impl Client {
    #[inline]
    pub fn connect<S: AsRef<str>>(
        transport: &Transport,
        framer: Framer,
        addr: S,
        timeout: Duration,
    ) -> Result<Self, Error> {
        let endpoint = transport
            .connect(addr, timeout)
            .map_err(Error::TransportError)?;

        Ok(Self(Peer::new(framer, endpoint)))
    }
}

impl Deref for Client {
    type Target = Peer;

    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl DerefMut for Client {
    fn deref_mut(&mut self) -> &mut Self::Target {
        &mut self.0
    }
}
